package com.share.common.rabbit.config;

import com.alibaba.fastjson2.JSON;
import com.share.common.rabbit.domain.SelfCorrelationData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * mq 配置类 开启发送者确认机制
 * 实现了 ApplicationListener<ApplicationReadyEvent> 接口 Spring 应用启动完成后执行初始化操作
 */
@Component
@Slf4j
public class RabbitInitConfig implements ApplicationListener<ApplicationReadyEvent> {

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private RedisTemplate redisTemplate;

    //在 Spring 应用启动完成后触发
    @Override
    public void onApplicationEvent(ApplicationReadyEvent event) {
        this.setupCallbacks();
    }


    private void setupCallbacks() {


        /**
         * 发送者确认机制，确认消息是否成功发送到指定交换机
         * 只确认消息是否到路由器中，不关心是否到队列中，只要成功发送到交换机中就返回ack
         *
         * @param correlation 相关数据  非消息本身业务数据
         * @param ack         返回ack说明成功发送 nack说明发送失败
         * @param reason      如果发送消息到交换器失败，错误原因
         */
        this.rabbitTemplate.setConfirmCallback((correlationData, ack, reason) -> {
            if (ack) {
                //消息发送到交换器成功  ack
                log.info("消息发送到Exchange成功：{}", correlationData);
            } else {
                //消息发送到交换器失败  nack
                log.error("消息发送到Exchange失败：{}", reason);

                //进行重发消息
                this.retrySendMsg(correlationData);
            }
        });

        /**
         * 确认消息是否成功路由到 Queue（队列）。当消息成功发送到 Exchange 但未能正确路由到队列时触发。
         * 如交换机没有绑定的队列，消息停留在交换机时会触发
         * 如果正确到达队列不执行
         */
        this.rabbitTemplate.setReturnsCallback(returned -> {
            log.error("消息未到达队列，消息: {}, 返回码: {}, 返回文本: {}, 交换机/路由键: {}/{}",
                    returned.getMessage(), returned.getReplyCode(), returned.getReplyText(),
                    returned.getExchange(), returned.getRoutingKey());

            //消息未正确路由到队列时，进行重发消息
            //从returned中获取消息的uuid 路由失败时，消息id会自动注入到该消息头中
            String messageId = returned.getMessage().getMessageProperties().getHeader("spring_returned_message_correlation");

            //从redis中读取第一次发送消息时 存储的数据
            String correlationDataStr = (String) redisTemplate.opsForValue().get(messageId);

            //将字符串转为 自定义消息对象类型后重发
            SelfCorrelationData selfCorrelationData = JSON.parseObject(correlationDataStr, SelfCorrelationData.class);

            //todo 方式一:如果不考虑延迟消息重发 直接返回
            if (selfCorrelationData.isDelay()) {
                return;
            }

            this.retrySendMsg(selfCorrelationData);

        });
    }

    //消息重发
    private void retrySendMsg(CorrelationData correlationData) {

        log.info("出现异常,重发消息！");

        //转换消息类型
        SelfCorrelationData selfCorrelationData = (SelfCorrelationData) correlationData;

        //可重发消息的次数
        int retryCount = selfCorrelationData.getRetryCount();
        if (retryCount >= 3) {
            //超过最大重试次数
            log.error("超过最大重试次数，请联系管理员");
            return;
        }

        //统计重发次数+1 并写入selfCorrelationData
        retryCount++;
        selfCorrelationData.setRetryCount(retryCount);

        //将redis中的消息数据进行更新，更新重发次数 重置过期时间
        redisTemplate.opsForValue().set(selfCorrelationData.getId(), JSON.toJSONString(selfCorrelationData), 10, TimeUnit.MINUTES);

        //重发消息 判断是普通消息还是延迟消息
        //todo 方式二：如果是延迟消息，依然需要设置消息延迟时间
        if (selfCorrelationData.isDelay()) {
            //延迟消息
            rabbitTemplate.convertAndSend(
                    selfCorrelationData.getExchange(),
                    selfCorrelationData.getRoutingKey(),
                    selfCorrelationData.getMessage(),
                    message -> {
                        message.getMessageProperties().setDelay(selfCorrelationData.getDelayTime() * 1000);
                        return message;
                    },
                    selfCorrelationData);
        } else {
            //普通消息
            // 重发消息到Rabbit   参数：交换机 routing key 消息数据 自定义回调消息
            rabbitTemplate.convertAndSend(selfCorrelationData.getExchange(), selfCorrelationData.getRoutingKey(),
                    selfCorrelationData.getMessage(), selfCorrelationData);
        }

    }

}
